package com.study.chapter09;

import com.study.chapter05.source.ClickSource;
import com.study.entity.Event;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.time.Duration;

/**
 * @Description:
 * @Author: LiuQun
 * @Date: 2022/8/15 22:20
 */
public class PeriodicPvExampleTest {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        SingleOutputStreamOperator<Event> stream = env.addSource(new ClickSource())
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ZERO)
                                .withTimestampAssigner((event, l) -> event.timestamp)
                );
        stream.print("====>");

        stream.keyBy(event -> event.user)
                .process(new PeriodicPvResult())
                .print();

        env.execute();
    }
    public static class PeriodicPvResult extends KeyedProcessFunction<String,Event,String>{
        //定义状态，保存当前pv统计值，以及有没有定时器
        ValueState<Long> countState;
        ValueState<Long> timerTsState;


        @Override
        public void open(Configuration parameters) throws Exception {
            countState = getRuntimeContext().getState(new ValueStateDescriptor<Long>("count-",Long.class));
            timerTsState = getRuntimeContext().getState(new ValueStateDescriptor<Long>("timer-ts",Long.class));

        }

        @Override
        public void processElement(Event value, Context ctx, Collector<String> out) throws Exception {
            //每来一条数据，就更新count的值(之前已经根据用户进行分组了)
            Long count = countState.value();
            countState.update(count == null ? 1 : count+1);

            //若没有注册过定时器，则注册一个定时器
            if (timerTsState.value() == null){
                ctx.timerService().registerEventTimeTimer(value.timestamp + 10*1000L);
                //更新时间戳
                timerTsState.update(value.timestamp + 10*1000L);
            }
        }

        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
            //定时器触发，输出一次结果
            out.collect(ctx.getCurrentKey()+"==>pv："+countState.value());
            //清空状态
            timerTsState.clear();
            //重新注册一个定时器
            ctx.timerService().registerEventTimeTimer(timestamp + 10*1000L);
            timerTsState.update(timestamp + 10*1000L);
        }
    }
}
